<?php

require_once('logger.php');


///////////////////////////////////////////////////////////////////////

function    makeTCPAddress($host, $port)
{
    $address = sprintf("tcp://%s:%s", $host, $port);
    return $address;
}

class    JuTCPChannel   extends Jujucell
{
    const RECORD_END_MARK = "\n";
    const RECORD_END_MARK_SIZE = 1;
    
    const RECV_BUF_SIZE   = 65536;
    
    private $recv_buf;    # for $this->recv() 
    
    private $context;
    private $jucore;
    
    private $logger;
    
    public function __construct($socket, $address=NULL, $context=NULL, $jucore=NULL)
    {
        $this->logger = getLogger();
        
        if (empty($address)) {
            socket_getpeername($socket, $addr_, $port_);
            $address = sprintf("tcp://%s:%s", $addr_, $port_);
            $this->logger->info(sprintf("channel to %s established.", $address));
        }
        
        parent::__construct($socket, $address);
        
        $this->socket = $socket;
        $this->recv_buf = '';
        
        $this->context = $context;
        $this->jucore = $jucore;

        if ($this->jucore) {
            $this->jucore->addChannel($this);
        }
        if ($this->context) {
            $this->context->scheduler->registerChannel($address, $this);
        }
    }
    
    public function OnRecvRecord($data)
    {
        $this->context->scheduler->dispatch($this, $data);
    }
    public function OnPreSendRecord()
    {
        $this->context->scheduler->OnPreSend($this);
    }
    public function OnPostSendRecord()
    {
        $this->context->scheduler->OnPostSend($this);
    }
    public function OnConnectionLost()
    {
        $this->context->scheduler->OnConnectionLost($this);
    }
    
    ///////////////////////////////////////////////////////////////////
    
    public function sendRecord($data)
    {
        $send_buf = $data;
        while (strlen($send_buf) > 0) {
            $ret = socket_write($this->socket, $send_buf);
            if (FALSE === $ret) {
                $errno = socket_last_error($this->socket);
                //if ((111 == $errno) || (104 == $errno)) { do something }
                $msg = sprintf('Socket WRITE Error -- [%d] %s', $errno, socket_strerror($errno));
                throw new Exception($msg, $errno);
            }
            $send_buf = substr($send_buf, $ret);
            //printf("SEND size=$ret\n");
        }
    }
    
    public function recvRecord(&$data)
    {
        $pos_scan = strlen($this->recv_buf) - self::RECORD_END_MARK_SIZE;
        if ($pos_scan < 0) {
            $pos_scan = 0;
        }
        while(true) {
            $pos = strpos($this->recv_buf, self::RECORD_END_MARK, $pos_scan);
            if (FALSE === $pos) {
                //$ret = socket_read($this->socket, self::RECV_BUF_SIZE);
                $ret = socket_read($this->socket, self::RECV_BUF_SIZE, PHP_NORMAL_READ);
                if (false === $ret) {
                    $errno = socket_last_error($this->socket);
                    //if ((111 == $errno) || (104 == $errno)) { do something }
                    $msg = sprintf('Socket READ Error -- [%d] %s', $errno, socket_strerror($errno));
                    throw new Exception($msg, $errno);
                }
                
                $pos_scan = strlen($this->recv_buf) - self::RECORD_END_MARK_SIZE;
                if ($pos_scan < 0) {
                    $pos_scan = 0;
                }
                $this->recv_buf = join('', array($this->recv_buf, $ret));
                //$this->recv_buf = $this->recv_buf . $ret;
            }
            else {
                $data = substr($this->recv_buf, 0, $pos);
                $this->recv_buf = substr($this->recv_buf, $pos + self::RECORD_END_MARK_SIZE );

                break;
            }
        }
        //var_dump($data);
        return strlen($data);
    }
}    // class JuTCPChannel 


class   JuTCPServer extends Jujucell
{
    public  $port;
    
    private $context;
    private $jucore;
    
    private $logger;
    
    public function __construct($port, $context=None, $jucore=None)
    {
        $this->logger = getLogger();
        
        $sock = socket_create_listen($port);
        if (false === $sock) {
            $errno = socket_last_error();
            $msg = sprintf("Error Creating Listen Socket on port %s -- [%d] %s", $port, $errno, socket_strerror($errno));
            throw new Exception($msg, $errno);
        }
        
        socket_getsockname($sock, $addr_, $port_);
        $address = sprintf("tcp://%s:%s", $addr_, $port_);
        $this->logger->info(sprintf('listening on %s ...', $address));
        parent::__construct($sock, $address);
        
        $this->port = $port;
        $this->context = $context;
        $this->jucore = $jucore;
        
        if ($this->jucore) {
            $this->jucore->addChannel($this);
        }
    }
    
    public function handleReadEvent()
    {
        $client_sock = socket_accept($this->socket);
        socket_getpeername($client_sock, $addr_, $port_);
        $client_address = sprintf("tcp://%s:%s", $addr_, $port_);
        $this->logger->info(sprintf("client %s connected.", $client_address));
        $channel = new JuTCPChannel($client_sock, $client_address, $this->context, $this->jucore);
    }
    
}   // class JuTCPServer

///////////////////////////////////////////////////////////////////////
//

class    ExitException extends Exception
{
}    // class ExitException

class    NotImplementedError extends Exception
{
}    // class NotImplementedError

class    RuntimeError extends Exception
{
}    // class RuntimeError


class    Jujucell
{
    const RECORD_END_MARK = "\n";
    const RECORD_END_MARK_SIZE = 1;
    
    const RECV_BUF_SIZE   = 65536;

    public  $socket;
    public  $address;
    private $connected;
    
    private $in_buf;    // for handleReadEvent()
    private $pos_scan;  
    
    private $out_buf;
    private $producer_fifo;
    
    private $last_error;

    private $logger;
    
    public function __construct($socket, $address)
    {
        $this->socket = $socket;
        $this->connected = true;
        $this->address = $address;
        
        $this->in_buf = '';
        $this->pos_scan = 0;
        
        $this->out_buf = '';
        $this->producer_fifo = new Fifo();
        
        $this->logger = getLogger();
    }
    
    public function OnRecvRecord($data)
    {
        throw new NotImplementedError();
    }
    public function OnPreSendRecord()
    {
        throw new NotImplementedError();
    }
    public function OnPostSendRecord()
    {
        throw new NotImplementedError();
    }
    public function OnConnectionLost()
    {
        throw new NotImplementedError();
    }
    
    
    public function push($data)
    {
        if (empty($data)) {
            throw new Exception('Empty Data to be pushed.');
        }
        
        $producer = new SimpleProducer($data);
        $this->producer_fifo->push($producer);
    }
    
    public function countProducers()
    {
        return $this->producer_fifo->count();
    }
    
    ///////////////////////////////////////////////////////////////////
    
    public function handleReadEvent()
    {
        //$ret = socket_read($this->socket, self::RECV_BUF_SIZE);
        $ret = @socket_read($this->socket, self::RECV_BUF_SIZE, PHP_NORMAL_READ);
        if (false === $ret) {
            $this->logger->debug(sprintf('%s address:%s', __FUNCTION__, $this->address), __FILE__, __LINE__);
            $errno = socket_last_error($this->socket);
            //if ((111 == $errno) || (104 == $errno)) { do something }
            $msg = sprintf('Socket %s READ Error -- [%d] %s', $this->address, $errno, socket_strerror($errno));
            $this->logger->error($msg, __FILE__, __LINE__);
            throw new ExitException($msg, $errno);
        }
        elseif (empty($ret)) {
            $msg = sprintf('Socket %s READ Broke', $this->address);
            throw new ExitException($msg);
        }
        
        //$this->in_buf = join('', array($this->in_buf, $ret));
        $this->in_buf = $this->in_buf . $ret;
        
        while(true) {
            $pos = strpos($this->in_buf, self::RECORD_END_MARK, $this->pos_scan);
            if (FALSE === $pos) {
                $this->pos_scan = strlen($this->in_buf) - self::RECORD_END_MARK_SIZE;
                if ($this->pos_scan < 1) {
                    $this->pos_scan = 0;
                }
                break;
            }
            else {
                $data = substr($this->in_buf, 0, $pos);
                $this->in_buf = substr($this->in_buf, $pos + self::RECORD_END_MARK_SIZE );
                $this->pos_scan = 0;    // scan from head, next time

                // callback data-handler
                $this->OnRecvRecord($data);
            }
        }
    }   // function handleReadEvent()
    
    public function handleWriteEvent()
    {
        $this->sendData();
    }
    
    public function handleExceptionEvent()
    {
        echo __FUNCTION__,__FILE__, __LINE__,"\n";
    }
    public function handleExitException()
    {
        $this->connected = false;
        $this->OnConnectionLost();
    }
    
    public function handleError()
    {
    }
    
    public function readable()
    {
        if ($this->connected) {
            return true;
        }
        else {
            return false;
        }
    }
    
    public function writable()
    {
        if ($this->connected) {
            if ($this->producer_fifo->count() > 0) {
                return true;
            }
        }
        return false;
    }
    
    ///////////////////////////////////////////////////////////////////
    
    private function sendData()
    {
        if (empty($this->out_buf)) {
            $this->refillOutBuffer();
            if (empty($this->out_buf)) {
                return false;
            }
        }
        
        $ret = socket_write($this->socket, $this->out_buf);
        if (FALSE === $ret) {
            $errno = socket_last_error($this->socket);
            //if ((111 == $errno) || (104 == $errno)) { do something }
            $msg = sprintf('Socket WRITE Error -- [%d] %s', $errno, socket_strerror($errno));
            throw new Exception($msg, $errno);
        }
        $this->out_buf = substr($this->out_buf, $ret);

        return true;
    }
    
    private function refillOutBuffer()
    {
        while (true) {
            if ($this->producer_fifo->count() < 1) {
                return false;
            }
            
            $producer = $this->producer_fifo->first();
            $data = $producer->more();
            if (empty($data)) {
                $this->producer_fifo->pop();
                $this->OnPostSendRecord();
                continue;
            }
            else {
                $this->OnPreSendRecord();
                $this->out_buf = join('', array($this->out_buf, $data));
                return true;
            }
        }
        
        return false;
    }
    
}    // class Jujucell

class   SimpleProducer
{
    private $data;
    private $buffer_size;
    
    public function __construct($data, $buffer_size=65536) 
    {
        $this->data = $data;
        $this->buffer_size = $buffer_size;
    }
    
    public function more()
    {
        $result = $this->data;
        $this->data = '';
        
        return $result;
    }
}   // class SimpleProducer

class   Fifo
{
    private $list;
    
    public function __construct()
    {
        $this->list = array();
    }
    
    public function count()
    {
        return count($this->list);
    }
    
    public function isEmpty()
    {
        return ($this->count() < 1);
    }
    
    public function first()
    {
        if ($this->isEmpty()) {
            return NULL;
        }
        return  $this->list[0];
    }
    
    public function push($obj)
    {
        array_push($this->list, $obj);
    }
    
    public function pop()
    {
        if ($this->count() > 0) {
            $result = array_shift($this->list);
            return $result;
        }
        else {
            return NULL;
        }
    }
}   // class Fifo


///////////////////////////////////////////////////////////////////////
//

class    Jujucore
{
    private $map;
    
    public function __construct()
    {
        $this->map = array();
    }
    
    public function addChannel($channel)
    {
        $sock = $channel->socket;
        if (isset($this->map[ $sock ])) {
            return true;
        }
        $this->map[ $sock ] = $channel;
    }
    
    public function delChannel($channel)
    {
        $sock = $channel->socket;
        if (isset($this->map[ $sock ])) {
            unset($this->map[ $sock ]);
        }
        return true;
    }
    
    public function loop($timeout=10, $count=0)
    {
        if ($timeout > 0) {
            $tv_sec = floor($timeout);
            $tv_usec = floor(($timeout - $tv_sec) * 1000000);
        }
        else {
            $tv_sec = 0;
            $tv_usec = 0;
        }
        $count_ = 0;
        
        while( count($this->map) > 0 ) {
            $r = array();
            $w = array();
            $e = array();
            
            foreach ($this->map as $sock => $obj) {
                $exceptable = false;
                if ($obj->readable()) {
                    $r[] = $obj->socket;
                    $exceptable = true;
                }
                if ($obj->writable()) {
                    $w[] = $obj->socket;
                    $exceptable = true;
                }
                if ($exceptable) {
                    $e[] = $obj->socket;
                }
            }
            
            $ret = socket_select($r, $w, $e, $tv_sec, $tv_usec);
            if (FALSE === $ret) {
                $errno = socket_last_error();
                $msg = sprintf('Socket SELECT Error -- [%d] %s', $errno, socket_strerror($errno));
                throw new Exception($msg, $errno);
            }
            elseif (0 === $ret) {
                return 0;
            }
            
            // handle exception
            foreach ($e as $sock) {
                $obj = $this->map[ $sock ];
                try {
                    $obj->handleExceptionEvent();
                }
                catch(ExitException $ex) {
                    unset($this->map[ $sock ]);
                }
                catch(Exception $ex) {
                    $obj->handleError();
                }
            }
                        
            // handle reading
            foreach ($r as $sock) {
                $obj = $this->map[ $sock ];
                try {
                    $obj->handleReadEvent();
                }
                catch(ExitException $ex) {
                    $obj->handleExitException();
                    unset($obj);
                    unset($this->map[ $sock ]);
                }
                catch(Exception $ex) {
                    $obj->handleError();
                }
            }
            
            // handle writing            
            foreach ($w as $sock) {
                $obj = $this->map[ $sock ];
                try {
                    $obj->handleWriteEvent();
                }
                catch(ExitException $ex) {
                    unset($this->map[ $sock ]);
                }
                catch(Exception $ex) {
                    $obj->handleError();
                }
            }
            
            // count loop 
            if ($count > 0) {
                ++$count_;
                if ($count_ >= $count) {
                    break;
                }
            }
        } // while(true)
    }    // loop()
    
}    // class Jujucore

